package com.aotain.jupiter.flink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;


public class FlinkTest {
	public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        String dbJson = "{\"driverClassName\":\"com.mysql.jdbc.Driver\","
    			+ "\"maxActive\":\"200\",\"maxIdle\":\"50\",\"maxWait\":\"10000\","
    			+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
    			+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:mysql://192.168.5.97:3306/SDS\""
    			+ ",\"userName\":\"sds\""
    			+ ",\"validationQuery\":\"SELECT SYSDATE()\"}";
		

        //Flink 作为 client 连接 server，server 发送数据过来
//        DataStream<Tuple2<String,String>> dataStream = env.socketTextStream("192.168.5.96", 9966)
//				.map(new SourceDataStat(dbJson));
        DataStream<Tuple2<String,Long>> dataStream = env.socketTextStream("192.168.5.96", 9999)
        		.map(new MapFunction<String, Tuple2<String,Long>>() {

					@Override
					public Tuple2<String,Long> map(String value) throws Exception {
						// TODO Auto-generated method stub
						String[] splits = value.split(" ");
						return new Tuple2<String,Long>(splits[0],Long.parseLong(splits[1]));
					}
        			
				});
        
        
//        dataStream.keyBy(0).timeWindow(Time.seconds(20)).reduce(new ReduceFunction<Tuple2<String,Long>>() {
//			
//			@Override
//			public Tuple2<String, Long> reduce(Tuple2<String, Long> value1,
//					Tuple2<String, Long> value2) throws Exception {
//				Tuple2<String,Long> retTuple  = new Tuple2<String,Long>();
//				// TODO Auto-generated method stub
//				if(	value2 != null) {
//					retTuple = value2;
//				}else if(value2 == null && value1 != null){
//					retTuple = value1;
//				}else {
//					System.out.println("#####AbnormalReduce###### stat is null");
//				}
//				return retTuple;
//			}
//		}).map(new MapFunction<Tuple2<String,Long>, Object>() {
//
//			@Override
//			public Object map(Tuple2<String, Long> value) throws Exception {
//				System.out.println(value.f1);
//				return null;
//			}
//		});
//        dataStream.keyBy(0).timeWindow(Time.minutes(20), Time.minutes(1))
//				.apply(new WindowFunction<Tuple2<String,Long>, String, Tuple, TimeWindow>() {
//
//					@Override
//					public void apply(Tuple key, TimeWindow window,
//							Iterable<Tuple2<String, Long>> input,
//							Collector<String> out) throws Exception {
//						// TODO Auto-generated method stub
//						
//					}
//				});
        
        //dataStream.flatMap(flatMapper)
        
        		//.flatMap(new IPFlowStat())
//        		.keyBy(0)
//        		.broadcast()
//        		.reduce(reducer)
//        		.timeWindow(Time.seconds(5))
//        		.sum(1);

//        dataStream.print();
        env.execute("Socket Window WordCount");
    }
}
